package com.aotain.jupiter.flink;

import java.util.Properties;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import com.aotain.jupiter.util.FilterNullEnum;


public class FlinkKafkaSource {

	public static void main(String [] args) throws Exception {

		//		String dbJson = "{\"driverClassName\":\"com.mysql.jdbc.Driver\","
		//    			+ "\"maxActive\":\"200\",\"maxIdle\":\"50\",\"maxWait\":\"10000\","
		//    			+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
		//    			+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:mysql://192.168.5.97:3306/SDS\""
		//    			+ ",\"userName\":\"sds\""
		//    			+ ",\"validationQuery\":\"SELECT SYSDATE()\"}";
		String dbJson_oracle ="{\"driverClassName\":\"oracle.jdbc.driver.OracleDriver\","
				+ "\"maxActive\":\"200\",\"maxIdle\":\"50\",\"maxWait\":\"10000\","
				+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
				+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:oracle:thin:@192.168.50.155:1521:bsmp\","
				+ "\"userName\":\"sds\",\"validationQuery\":\"select sysdate from dual\"}";
		String filterIpSql = "SELECT DISTINCT A.IP FROM SDS_SERVERINFO A JOIN SDS_WEBSITEINFO B ON A.SITEID = B.SITEID WHERE B.MONITORSTATUS = 1";
		String whiteIpSql = " SELECT * FROM SDS_WHITELIST";

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		Properties properties = new Properties();
		properties.setProperty("bootstrap.servers", "192.168.5.95:9092,192.168.5.96:9092,192.168.5.97:9092");
		// only required for Kafka 0.8
		properties.setProperty("zookeeper.connect", "192.168.5.97:2181,192.168.5.98:2181,192.168.5.95:2181/kafka_2");
		properties.setProperty("group.id", "flink");

		//		DataStream<String> stream = env
		//				.addSource(new FlinkKafkaConsumer08<>("udii", new SimpleStringSchema(), properties))
		//				.map(new MapFunction<String, String>() {
		//
		//					@Override
		//					public String map(String value) throws Exception {
		//						// TODO Auto-generated method stub
		//						return value;
		//					}
		//					
		//				});
		//		stream.print();

		//		DataStream<Tuple2<String,String>> stream = env
		//				.addSource(new FlinkKafkaConsumer08<>("udii", new SimpleStringSchema(), properties))
		//				.map(new SourceDataStat(dbJson));

		//		DataStream<Tuple2<String,String>> stream = env
		//				.addSource(new FlinkKafkaConsumer08<>("udii", new SimpleStringSchema(), properties))
		//				.map(new SourceDataStat(dbJson_oracle));
		//		stream.map(new WhiteIpMap("SELECT DISTINCT A.IP FROM SDS_SERVERINFO A JOIN SDS_WEBSITEINFO B ON A.SITEID = B.SITEID WHERE B.MONITORSTATUS = 1", "{\"driverClassName\":\"oracle.jdbc.driver.OracleDriver\","
		//						+ "\"maxActive\":\"200\",\"maxIdle\":\"50\",\"maxWait\":\"10000\","
		//						+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
		//						+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:oracle:thin:@192.168.50.155:1521:bsmp\","
		//						+ "\"userName\":\"sds\",\"validationQuery\":\"select sysdate from dual\"}"));

		DataStream<String> stream = env
				.addSource(new FlinkKafkaConsumer08<>("udii", new SimpleStringSchema(), properties))
				.map(new SourceDataFilterMap(filterIpSql,whiteIpSql, dbJson_oracle));
		SplitStream<String> splitStream = stream.split(new SourceDataSelector());
		DataStream<Tuple2<String, String>> dipStream = splitStream.select("DipStream")
				.map(new DipSoucreStreamStat(dbJson_oracle))
				.filter(new FilterFunction<Tuple2<String,String>>() {

					@Override
					public boolean filter(Tuple2<String, String> value) throws Exception {
						// TODO Auto-generated method stub
						if(value.f0 == null) {
							return false;
						}else {
							return true;
						}
					}
				})
				.map(new MapFunction<Tuple2<String,String>, Tuple2<String,String>>() {

					@Override
					public Tuple2<String, String> map(
							Tuple2<String, String> value) throws Exception {
						// TODO Auto-generated method stub
						System.out.println("######DIPStream" + value.f1);
						return null;
					}

				})
				.setParallelism(8);
		DataStream<Tuple2<String, String>> sipStream = splitStream.select("SipStream").
				map(new DipSoucreStreamStat(dbJson_oracle))
				.filter(new FilterFunction<Tuple2<String,String>>() {

					@Override
					public boolean filter(Tuple2<String, String> value) throws Exception {
						// TODO Auto-generated method stub
						if(value.f0 == null) {
							return false;
						}else {
							return true;
						}
					}
				})
				.map(new MapFunction<Tuple2<String,String>, Tuple2<String,String>>() {

					@Override
					public Tuple2<String, String> map(
							Tuple2<String, String> value) throws Exception {
						// TODO Auto-generated method stub
						System.out.println("######SIPStream" + value.f1);
						return null;
					}

				})
				.setParallelism(8);


		//--------------异常流量检测------------/
		//		WindowedStream<Tuple2<String,AbnormalStatTuple>,Tuple,TimeWindow> abnStream = stream.map(new AbnormalStatMap())
		//				.filter(new FilterNullObject<AbnormalStatTuple>(FilterNullEnum.F1ISNOTNULL)).keyBy(0).timeWindow(Time.seconds(10));
		//		abnStream.reduce(new AbnormalReduce()).flatMap(new AbnormalCheckFlink());

		//--------------扫描攻击监测-------------
		//		WindowedStream<Tuple2<String,ScanAttackTuple>,Tuple,TimeWindow> scanAttStream = stream.map(new ScanAttackStatMap())
		//				.filter(new FilterNullObject<ScanAttackTuple>(FilterNullEnum.F0ISNOTNULL)).keyBy(0).timeWindow(Time.minutes(1));
		//		scanAttStream.reduce(new ScanAttackReduce()).flatMap(new ScanAttackFlink());

		//		DataStream<Tuple2<String, Long>> stream  = env
		//				.addSource(new FlinkKafkaConsumer08<>("udii", new SimpleStringSchema(), properties))
		//				.map(new IPFlowStat());
		//		stream.print();

		//		abnStream.reduce(new AbnormalReduce()).map( new AbnormalReadyCheckMap()).flatMap(new AbnormalCheckFlink());
		//		WindowedStream<Tuple2<String, CCStatTuple>, Tuple, TimeWindow> winCC 
		//		= dataStream.map(new URLStatMap()).filter(new FilterFunction<Tuple2<String,CCStatTuple>> (){
		//			/**
		//			 * 
		//			 */
		//			private static final long serialVersionUID = 3148911221235593226L;
		//			@Override
		//			public boolean filter(Tuple2<String, CCStatTuple> arg0)
		//					throws Exception {
		//				// TODO Auto-generated method stub
		//				if(arg0.f1 == null) {
		//					return false;
		//				} else {
		//					return true;
		//				}
		//			}
		//			
		//		}).keyBy(0).timeWindow(Time.minutes(1));
		//	
		//	
		//	winCC.reduce(new URLStatReduce()).flatMap(new CCDetectFlink());

		env.execute("FlinkSourceStat");
	}
}
